package com.atguigu.flink.streamapi.transform;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.sql.Connection;

/**
 * Created by Smexy on 2022/11/21
 *
 *  map: 输入1，必须输出1
 *
 *  在运算时，需要读取数据库。
 *      希望一个Task，创建一个连接即可。这个Task收到的所有数据，都共享这个连接。
 *              Task创建时，新建一个连接，共享。只做一次
 *              Task关闭时，关闭连接。只做一次
 *
 *       以上需求，需要使用
 *               RichMapFunction<IN, OUT> extends AbstractRichFunction
 *                  AbstractRichFunction提供了生命周期方法(在Task的整个生命中执行一次)
 *                          open:  task创建时执行一次
 *                          close: task关闭时执行一次
 *
 *    --------------------------
 *      所有的算子，都提供了 RichXxxFunction,和RichMapFunction类似！
 */
public class Demo2_RichMap
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //env.setParallelism(1);

        env.fromElements(1,2,3)
           .map(new RichMapFunction<Integer, Integer>()
           {
               Connection connection;
               @Override
               public void open(Configuration parameters) throws Exception {
                   connection = null;
                   System.out.println("新建连接,读数据库");
               }

               @Override
               public void close() throws Exception {
                   System.out.println("关闭连接");
               }

               @Override
               public Integer map(Integer value) throws Exception {
                   System.out.println(connection);
                   return value + 1;
               }
           })    //默认取本地环境的cpu数作为 并行度
           .print();

        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }

    }
}
